package com.credithc.eagleeye.plugins.rabbitmq;
import com.credithc.eagleeye.agent.factory.EagleEyeAgentFactory;
import com.credithc.eagleeye.agent.util.HostUtil;
import com.credithc.eagleeye.constants.RpcTypeEnum;
import com.credithc.eagleeye.exception.EagleEyeIgnoreException;
import com.credithc.eagleeye.thread.EagleEyeThreadContext;
import com.credithc.eagleeye.thread.TraceInfo;
import com.credithc.eagleeye.util.FastJsonConvert;
import com.credithc.eagleeye.util.MixUtil;
import com.credithc.eagleeye.util.ThrowableToStringArray;
import com.credithc.eagleeye.vo.RpcTraceInfoVo;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.io.UnsupportedEncodingException;
import java.util.Date;
import java.util.Map;

import static com.credithc.eagleeye.constants.EagleEyeConstant.HC_RPC_ID;
import static com.credithc.eagleeye.constants.EagleEyeConstant.HC_TRACE_ID;
import static com.credithc.eagleeye.plugins.util.ClassUtil.rabbitCorrelationDataClass;
/**
 * 描 述: <br/>
 * 作 者: zhangjin<br/>
 * 创 建：2019年10月24日<br/>
 * 版 本：v1.0.0<br>
 * <p>
 * 历 史: (版本) 作者 时间 注释 <br/>
 */
@Slf4j
@Aspect
public class EagleEyeRabbitPublishAspect {
    private static int maxLength = 4000;

    public EagleEyeRabbitPublishAspect() {
        log.info("EagleEyeRabbitPublishAspect init..");
    }

    @Pointcut("execution(* org.springframework.amqp.rabbit.core.RabbitTemplate.convertAndSend(..)) ||" +
            "execution(* org.springframework.amqp.rabbit.core.RabbitTemplate.send(..)) ||" +
            "execution(* org.springframework.amqp.rabbit.core.RabbitTemplate.correlationConvertAndSend(..)) ||" +
            "execution(* org.springframework.amqp.rabbit.core.RabbitTemplate.convertSendAndReceive(..)) ||" +
            "execution(* org.springframework.amqp.rabbit.core.RabbitTemplate.sendAndReceive(..))")
    public void pointcut() {
    }

    @Around("pointcut()")
    public Object around(ProceedingJoinPoint joinPoint) throws Throwable {
        log.debug("EagleEyeRabbitPublishAspect around start...");
        Object resultObject=null;
        MethodSignature methodSignature = (MethodSignature) joinPoint.getSignature();
        String methodName = methodSignature.getMethod().getName();
        Class[] parameterTypes = methodSignature.getParameterTypes();
        Object[] args = joinPoint.getArgs();
        TraceInfo traceInfo = EagleEyeThreadContext.getTraceInfo();
        if (traceInfo == null) {
            traceInfo = new TraceInfo();
            EagleEyeThreadContext.init();
            EagleEyeThreadContext.putTraceInfo(traceInfo);
        } else {
            traceInfo.addSequenceNo();
        }
        Class correlationDataClass = rabbitCorrelationDataClass;


        RpcTraceInfoVo rpcTraceInfoVo = initRpcTraceInfoVo(traceInfo, methodName);
        if (args != null) {
            //convertAndSend已经由EagleEyeBeforePublishPostProcessor将header放进去了，这里只补全非Convert方法的
            if (StringUtils.equals(methodName, "send") || StringUtils.equals(methodName, "sendAndReceive")) {
                injectEagleEyeHeader(args, traceInfo);
            }
            if (StringUtils.equals(methodName, "send")) {
                if (parameterTypes.length == 1 && parameterTypes[0] == Message.class) {
                    //send(Message message)
                    fillRpcInfo(rpcTraceInfoVo, args[0], "", "");
                } else if (parameterTypes.length == 2 && parameterTypes[0] == String.class && parameterTypes[1] == Message.class) {
                    //send(String routingKey, Message message)
                    fillRpcInfo(rpcTraceInfoVo, args[1], "", args[0]);
                } else if (parameterTypes.length == 3 && parameterTypes[0] == String.class && parameterTypes[1] == String.class && parameterTypes[2] == Message.class) {
                    //send(final String exchange, final String routingKey, final Message message)
                    fillRpcInfo(rpcTraceInfoVo, args[2], args[0], args[1]);
                } else if (parameterTypes.length == 4 && parameterTypes[0] == String.class && parameterTypes[1] == String.class && parameterTypes[2] == Message.class && parameterTypes[3] == correlationDataClass) {
                    //send(final String exchange, final String routingKey,final Message message, @Nullable final CorrelationData correlationData)
                    fillRpcInfo(rpcTraceInfoVo, args[2], args[0], args[1]);
                }
            } else if (StringUtils.equals(methodName, "sendAndReceive")) {
                if (parameterTypes.length == 1 && parameterTypes[0] == Message.class) {
                    //sendAndReceive(final Message message)
                    fillRpcInfo(rpcTraceInfoVo, args[0]);
                } else if (parameterTypes.length == 2 && parameterTypes[0] == Message.class && parameterTypes[1] == correlationDataClass) {
                    //sendAndReceive(final Message message, @Nullable CorrelationData correlationData)
                    fillRpcInfo(rpcTraceInfoVo, args[0]);
                } else if (parameterTypes.length == 2 && parameterTypes[0] == String.class && parameterTypes[1] == Message.class) {
                    //sendAndReceive(final String routingKey, final Message message)
                    fillRpcInfo(rpcTraceInfoVo, args[1], "", args[0]);
                } else if (parameterTypes.length == 3 && parameterTypes[0] == String.class && parameterTypes[1] == Message.class && parameterTypes[2] == correlationDataClass) {
                    //sendAndReceive(final String routingKey, final Message message,@Nullable CorrelationData correlationData)
                    fillRpcInfo(rpcTraceInfoVo, args[1], "", args[0]);
                } else if (parameterTypes.length == 3 && parameterTypes[0] == String.class && parameterTypes[1] == String.class && parameterTypes[2] == Message.class) {
                    //sendAndReceive(final String exchange, final String routingKey, final Message message)
                    fillRpcInfo(rpcTraceInfoVo, args[2], args[0], args[1]);
                } else if (parameterTypes.length == 4 && parameterTypes[0] == String.class && parameterTypes[1] == String.class && parameterTypes[2] == Message.class && parameterTypes[3] == correlationDataClass) {
                    //sendAndReceive(final String exchange, final String routingKey, final Message message,@Nullable CorrelationData correlationData)
                    fillRpcInfo(rpcTraceInfoVo, args[2], args[0], args[1]);
                }
            } else if (StringUtils.equals(methodName, "convertAndSend")) {
                if (parameterTypes.length == 1 && parameterTypes[0] == Object.class) {
                    //convertAndSend(Object object)
                    fillRpcInfo(rpcTraceInfoVo, args[0]);
                } else if (parameterTypes.length == 2 && parameterTypes[0] == Object.class && parameterTypes[1] == MessagePostProcessor.class) {
                    //convertAndSend(Object message, MessagePostProcessor messagePostProcessor)
                    fillRpcInfo(rpcTraceInfoVo, args[0]);
                } else if (parameterTypes.length == 3 && parameterTypes[0] == Object.class && parameterTypes[1] == MessagePostProcessor.class && parameterTypes[2] == correlationDataClass) {
                    //convertAndSend(Object message, MessagePostProcessor messagePostProcessor,CorrelationData correlationData)
                    fillRpcInfo(rpcTraceInfoVo, args[0]);
                } else if (parameterTypes.length == 2 && parameterTypes[0] == String.class && parameterTypes[1] == Object.class) {
                    //convertAndSend(String routingKey, final Object object)
                    fillRpcInfo(rpcTraceInfoVo, args[1], "", args[0]);
                } else if (parameterTypes.length == 3 && parameterTypes[0] == String.class && parameterTypes[1] == Object.class && parameterTypes[2] == correlationDataClass) {
                    //convertAndSend(String routingKey, final Object object, CorrelationData correlationData)
                    fillRpcInfo(rpcTraceInfoVo, args[1], "", args[0]);
                } else if (parameterTypes.length == 3 && parameterTypes[0] == String.class && parameterTypes[1] == String.class && parameterTypes[2] == Object.class) {
                    //convertAndSend(String exchange, String routingKey, final Object object)
                    fillRpcInfo(rpcTraceInfoVo, args[2], args[0], args[1]);
                } else if (parameterTypes.length == 3 && parameterTypes[0] == String.class && parameterTypes[1] == Object.class && parameterTypes[2] == MessagePostProcessor.class) {
                    //convertAndSend(String routingKey, Object message, MessagePostProcessor messagePostProcessor)
                    fillRpcInfo(rpcTraceInfoVo, args[1], "", args[0]);
                } else if (parameterTypes.length == 4 && parameterTypes[0] == String.class && parameterTypes[1] == String.class && parameterTypes[2] == Object.class && parameterTypes[3] == correlationDataClass) {
                    //convertAndSend(String exchange, String routingKey, final Object object, @Nullable CorrelationData correlationData)
                    fillRpcInfo(rpcTraceInfoVo, args[2], args[0], args[1]);
                } else if (parameterTypes.length == 4 && parameterTypes[0] == String.class && parameterTypes[1] == Object.class && parameterTypes[2] == MessagePostProcessor.class && parameterTypes[3] == correlationDataClass) {
                    //convertAndSend(String routingKey, Object message, MessagePostProcessor messagePostProcessor, CorrelationData correlationData)
                    fillRpcInfo(rpcTraceInfoVo, args[1], "", args[0]);
                } else if (parameterTypes.length == 4 && parameterTypes[0] == String.class && parameterTypes[1] == String.class && parameterTypes[2] == Object.class && parameterTypes[3] == MessagePostProcessor.class) {
                    //convertAndSend(String exchange, String routingKey, final Object message, final MessagePostProcessor messagePostProcessor)
                    fillRpcInfo(rpcTraceInfoVo, args[2], args[0], args[1]);
                } else if (parameterTypes.length == 5 && parameterTypes[0] == String.class && parameterTypes[1] == String.class) {
                    //convertAndSend(String exchange, String routingKey, final Object message,final MessagePostProcessor messagePostProcessor,@Nullable CorrelationData correlationData)
                    fillRpcInfo(rpcTraceInfoVo, args[2], args[0], args[1]);
                }
            } else if (StringUtils.equals(methodName, "correlationConvertAndSend")) {
                if (parameterTypes.length == 2 && parameterTypes[0] == Object.class && parameterTypes[1] == correlationDataClass) {
                    //correlationConvertAndSend(Object object, CorrelationData correlationData)
                    fillRpcInfo(rpcTraceInfoVo, args[0]);
                }
            } else if (StringUtils.equals(methodName, "convertSendAndReceive")) {
                if (parameterTypes.length == 1 && parameterTypes[0] == Object.class) {
                    //convertSendAndReceive(final Object message)
                    fillRpcInfo(rpcTraceInfoVo, args[0]);
                } else if (parameterTypes.length == 2 && parameterTypes[0] == Object.class && parameterTypes[1] == correlationDataClass) {
                    //convertSendAndReceive(final Object message, @Nullable CorrelationData correlationData)
                    fillRpcInfo(rpcTraceInfoVo, args[0]);
                } else if (parameterTypes.length == 2 && parameterTypes[0] == Object.class && parameterTypes[1] == MessagePostProcessor.class) {
                    //convertSendAndReceive(final Object message, final MessagePostProcessor messagePostProcessor)
                    fillRpcInfo(rpcTraceInfoVo, args[0]);
                } else if (parameterTypes.length == 3 && parameterTypes[0] == Object.class && parameterTypes[1] == MessagePostProcessor.class && parameterTypes[2] == correlationDataClass) {
                    //convertSendAndReceive(final Object message, final MessagePostProcessor messagePostProcessor,@Nullable CorrelationData correlationData)
                    fillRpcInfo(rpcTraceInfoVo, args[0]);
                } else if (parameterTypes.length == 2 && parameterTypes[0] == String.class && parameterTypes[1] == Object.class) {
                    //convertSendAndReceive(final String routingKey, final Object message)
                    fillRpcInfo(rpcTraceInfoVo, args[1], "", args[0]);
                } else if (parameterTypes.length == 3 && parameterTypes[0] == String.class && parameterTypes[1] == Object.class && parameterTypes[2] == correlationDataClass) {
                    //convertSendAndReceive(final String routingKey, final Object message,@Nullable CorrelationData correlationData)
                    fillRpcInfo(rpcTraceInfoVo, args[1], "", args[0]);
                } else if (parameterTypes.length == 3 && parameterTypes[0] == String.class && parameterTypes[1] == String.class && parameterTypes[2] == Object.class) {
                    //convertSendAndReceive(final String exchange, final String routingKey, final Object message)
                    fillRpcInfo(rpcTraceInfoVo, args[2], args[0], args[1]);
                } else if (parameterTypes.length == 4 && parameterTypes[0] == String.class && parameterTypes[1] == String.class && parameterTypes[2] == Object.class && parameterTypes[3] == correlationDataClass) {
                    //convertSendAndReceive(final String exchange, final String routingKey, final Object message,@Nullable CorrelationData correlationData)
                    fillRpcInfo(rpcTraceInfoVo, args[2], args[0], args[1]);
                } else if (parameterTypes.length == 3 && parameterTypes[0] == String.class && parameterTypes[1] == Object.class && parameterTypes[2] == MessagePostProcessor.class) {
                    //convertSendAndReceive(final String routingKey, final Object message,final MessagePostProcessor messagePostProcessor)
                    fillRpcInfo(rpcTraceInfoVo, args[1], "", args[0]);
                } else if (parameterTypes.length == 4 && parameterTypes[0] == String.class && parameterTypes[1] == Object.class && parameterTypes[2] == MessagePostProcessor.class && parameterTypes[3] == correlationDataClass) {
                    //convertSendAndReceive(final String routingKey, final Object message, final MessagePostProcessor messagePostProcessor,@Nullable CorrelationData correlationData)
                    fillRpcInfo(rpcTraceInfoVo, args[1], "", args[0]);
                } else if (parameterTypes.length == 4 && parameterTypes[0] == String.class && parameterTypes[1] == String.class && parameterTypes[2] == Object.class && parameterTypes[3] == MessagePostProcessor.class) {
                    //convertSendAndReceive(final String exchange, final String routingKey, final Object message,final MessagePostProcessor messagePostProcessor)
                    fillRpcInfo(rpcTraceInfoVo, args[2], args[0], args[1]);
                } else if (parameterTypes.length == 5 && parameterTypes[0] == String.class && parameterTypes[1] == String.class && parameterTypes[2] == Object.class && parameterTypes[3] == MessagePostProcessor.class) {
                    //convertSendAndReceive(final String exchange, final String routingKey, final Object message,@Nullable final MessagePostProcessor messagePostProcessor,@Nullable final CorrelationData correlationData)
                    fillRpcInfo(rpcTraceInfoVo, args[2], args[0], args[1]);
                }
            }
        }

        //执行调用
        long beginTime = System.currentTimeMillis();
        try {
            resultObject = joinPoint.proceed();
            if(resultObject!=null) {
                rpcTraceInfoVo.setResponseJson(FastJsonConvert.convertObjectToJSON(resultObject));
            }
            rpcTraceInfoVo.setRunTime(System.currentTimeMillis()-beginTime);
            rpcTraceInfoVo.setResult("OK");
        } catch (Exception ex) {
            if (!(ex instanceof EagleEyeIgnoreException)) {
                //ERROR
                rpcTraceInfoVo.setResult("ERROR");
                rpcTraceInfoVo.setRunTime(System.currentTimeMillis()-beginTime);
                //异常的结果
                String[] traceStr = ThrowableToStringArray.convert(ex);
                StringBuilder builder = new StringBuilder();
                for(String trace : traceStr) {
                    builder.append(trace);
                    builder.append("\n");
                }
                rpcTraceInfoVo.setResponseJson(builder.toString());
            } else {
                //OK
                rpcTraceInfoVo.setResult("OK");
            }
            throw ex;
        } finally {
            EagleEyeAgentFactory.getInstance().send2EagleEye(rpcTraceInfoVo);
        }
        log.debug("EagleEyeRabbitPublishAspect around end...");
        return resultObject;
    }

    private void injectEagleEyeHeader(Object[] args, TraceInfo traceInfo) {
        for(Object arg : args) {
            if (arg instanceof Message) {
                Message message = (Message) arg;
                MessageProperties messageProperties = message.getMessageProperties();
                Map<String, Object> headers = messageProperties.getHeaders();
                headers.put(HC_TRACE_ID, traceInfo.getTraceId());
                headers.put(HC_RPC_ID, traceInfo.getRpcId());
                break;
            }
        }
    }

    private String fixRequestJson(String messageStr) {
        if (messageStr != null && messageStr.length() > maxLength) {
            return StringUtils.substring(messageStr, 0, maxLength) + "..";
        }
        return messageStr;
    }

    private void fillRpcInfo(RpcTraceInfoVo rpcTraceInfoVo, Object message) {
        fillRpcInfo(rpcTraceInfoVo, message, "", "");
    }

    private void fillRpcInfo(RpcTraceInfoVo rpcTraceInfoVo, Object message, Object exchange, Object routingKey) {
        StringBuilder requestJsonBuilder = new StringBuilder();
        requestJsonBuilder.append("exchange:" + exchange+"\n");
        requestJsonBuilder.append("routingKey:" + routingKey+"\n");
        requestJsonBuilder.append("body:" );
        if (message instanceof String) {
            requestJsonBuilder.append(fixRequestJson((String) message));
            rpcTraceInfoVo.setRequestSize(((String) message).length());
        } else if (message instanceof byte[]) {
            requestJsonBuilder.append("data type byte[]");
            rpcTraceInfoVo.setRequestSize(((byte[]) message).length);
        } else if (message instanceof Message) {
            Message messageRabbit = (Message) message;
            MessageProperties messageProperties = messageRabbit.getMessageProperties();
            rpcTraceInfoVo.setRequestSize(messageProperties.getContentLength());
            if (messageProperties.getContentType().equals(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)) {
                try {
                    requestJsonBuilder.append(fixRequestJson(new String(messageRabbit.getBody(), messageProperties.getContentEncoding())));
                } catch (UnsupportedEncodingException e) {
                }
            } else if (messageProperties.getContentType().equals(MessageProperties.CONTENT_TYPE_SERIALIZED_OBJECT)) {
                requestJsonBuilder.append("data type object");
            } else if (messageProperties.getContentType().equals(MessageProperties.CONTENT_TYPE_BYTES)) {
                requestJsonBuilder.append("data type byte[]");
            }
        } else {
            ByteArrayOutputStream output = new ByteArrayOutputStream();
            try {
                if ((message instanceof Serializable)) {
                    ObjectOutputStream objectOutputStream = new ObjectOutputStream(output);
                    objectOutputStream.writeObject(message);
                    objectOutputStream.flush();
                }
            } catch (IOException e) {
            }
            byte[] bytes = output.toByteArray();
            rpcTraceInfoVo.setRequestSize(bytes.length);
            if (bytes.length > maxLength) {
                requestJsonBuilder.append("data object too large ...");
            } else {
                requestJsonBuilder.append(FastJsonConvert.convertObjectToJSON(message));
            }
        }
        rpcTraceInfoVo.setRequestJson(requestJsonBuilder.toString());
    }

    private RpcTraceInfoVo initRpcTraceInfoVo(TraceInfo traceInfo, String methodName) {
        RpcTraceInfoVo rpcTraceInfoVo = new RpcTraceInfoVo();
        Date date = new Date();
        rpcTraceInfoVo.setRequestDateTime(MixUtil.ISO_DATETIME_TIME_ZONE_FORMAT_WITH_MILLIS.format(date));
        rpcTraceInfoVo.setTraceId(traceInfo.getTraceId());
        rpcTraceInfoVo.setRpcId(traceInfo.getRpcId());
        rpcTraceInfoVo.setRpcType(RpcTypeEnum.MQ.getRpcName());
        rpcTraceInfoVo.setServiceCategory("rabbitmq");
        rpcTraceInfoVo.setServiceName("rabbitmq");
        rpcTraceInfoVo.setMethodName(methodName);
        rpcTraceInfoVo.setClientHost(HostUtil.getIp());
        rpcTraceInfoVo.setServerHost("");
        return rpcTraceInfoVo;
    }
}
